package com.atguigu.hadoop.examples.mapreduce.dancing;


import com.google.common.base.Charsets;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.util.List;
import java.util.StringTokenizer;

public class DistributedPentomino extends Configured
        implements Tool
{
    private static final int PENT_DEPTH = 5;
    private static final int PENT_WIDTH = 9;
    private static final int PENT_HEIGHT = 10;
    private static final int DEFAULT_MAPS = 2000;

    private static long createInputDirectory(FileSystem fs, Path dir, Pentomino pent, int depth)
            throws IOException
    {
        fs.mkdirs(dir);
        List<int[]> splits = pent.getSplits(depth);
        Path input = new Path(dir, "part1");

        PrintWriter file = new PrintWriter(new OutputStreamWriter(new BufferedOutputStream(fs
                .create(input),
                65536), Charsets.UTF_8));
        for (int[] prefix : splits) {
            for (int i = 0; i < prefix.length; i++) {
                if (i != 0) {
                    file.print(',');
                }
                file.print(prefix[i]);
            }
            file.print('\n');
        }
        file.close();
        return fs.getFileStatus(input).getLen();
    }

    public static void main(String[] args)
            throws Exception
    {
        int res = ToolRunner.run(new Configuration(), new DistributedPentomino(), args);

        System.exit(res);
    }

    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        if (args.length == 0) {
            System.out.println("Usage: pentomino <output> [-depth #] [-height #] [-width #]");
            ToolRunner.printGenericCommandUsage(System.out);
            return 2;
        }

        int width = conf.getInt("mapreduce.pentomino.width", 9);
        int height = conf.getInt("mapreduce.pentomino.height", 10);
        int depth = conf.getInt("mapreduce.pentomino.depth", 5);
        for (int i = 0; i < args.length; i++) {
            if (args[i].equalsIgnoreCase("-depth"))
                depth = Integer.parseInt(args[(++i)].trim());
            else if (args[i].equalsIgnoreCase("-height"))
                height = Integer.parseInt(args[(++i)].trim());
            else if (args[i].equalsIgnoreCase("-width")) {
                width = Integer.parseInt(args[(++i)].trim());
            }

        }

        conf.setInt("mapreduce.pentomino.width", width);
        conf.setInt("mapreduce.pentomino.height", height);
        conf.setInt("mapreduce.pentomino.depth", depth);
        Class pentClass = conf.getClass("mapreduce.pentomino.class", OneSidedPentomino.class, Pentomino.class);

        int numMaps = conf.getInt("mapreduce.job.maps", 2000);
        Path output = new Path(args[0]);
        Path input = new Path(output + "_input");
        FileSystem fileSys = FileSystem.get(conf);
        try {
            Job job = Job.getInstance(conf);
            FileInputFormat.setInputPaths(job, new Path[] { input });
            FileOutputFormat.setOutputPath(job, output);
            job.setJarByClass(PentMap.class);

            job.setJobName("dancingElephant");
            Pentomino pent = (Pentomino)ReflectionUtils.newInstance(pentClass, conf);
            pent.initialize(width, height);
            long inputSize = createInputDirectory(fileSys, input, pent, depth);

            FileInputFormat.setMaxInputSplitSize(job, inputSize / numMaps);

            job.setOutputKeyClass(Text.class);

            job.setOutputValueClass(Text.class);

            job.setMapperClass(PentMap.class);
            job.setReducerClass(Reducer.class);

            job.setNumReduceTasks(1);

            return job.waitForCompletion(true) ? 0 : 1;
        } finally {
            fileSys.delete(input, true);
        }
    }

    public static class PentMap extends Mapper<WritableComparable<?>, Text, Text, Text>
    {
        private int width;
        private int height;
        private int depth;
        private Pentomino pent;
        private Text prefixString;
        private Mapper<WritableComparable<?>, Text, Text, Text>.Context context;

        public void map(WritableComparable<?> key, Text value, Mapper<WritableComparable<?>, Text, Text, Text>.Context context)
                throws IOException
        {
            this.prefixString = value;
            StringTokenizer itr = new StringTokenizer(this.prefixString.toString(), ",");
            int[] prefix = new int[this.depth];
            int idx = 0;
            while (itr.hasMoreTokens()) {
                String num = itr.nextToken();
                prefix[(idx++)] = Integer.parseInt(num);
            }
            this.pent.solve(prefix);
        }

        public void setup(Mapper<WritableComparable<?>, Text, Text, Text>.Context context)
        {
            this.context = context;
            Configuration conf = context.getConfiguration();
            this.depth = conf.getInt("mapreduce.pentomino.depth", 5);
            this.width = conf.getInt("mapreduce.pentomino.width", 9);
            this.height = conf.getInt("mapreduce.pentomino.height", 10);
            this.pent =
                    ((Pentomino)ReflectionUtils.newInstance(conf
                                    .getClass("mapreduce.pentomino.class", OneSidedPentomino.class),
                            conf));

            this.pent.initialize(this.width, this.height);
            this.pent.setPrinter(new SolutionCatcher());
        }

        class SolutionCatcher
                implements DancingLinks.SolutionAcceptor<Pentomino.ColumnName>
        {
            SolutionCatcher()
            {
            }

            public void solution(List<List<Pentomino.ColumnName>> answer)
            {
                String board = Pentomino.stringifySolution(DistributedPentomino.PentMap.this.width, DistributedPentomino.PentMap.this.height, answer);
                try {
                    DistributedPentomino.PentMap.this.context.write(DistributedPentomino.PentMap.this.prefixString, new Text("\n" + board));
                    DistributedPentomino.PentMap.this.context.getCounter(DistributedPentomino.PentMap.this.pent.getCategory(answer)).increment(1L);
                } catch (IOException e) {
                    System.err.println(StringUtils.stringifyException(e));
                } catch (InterruptedException ie) {
                    System.err.println(StringUtils.stringifyException(ie));
                }
            }
        }
    }
}